package com.matrix.draft.kafka.common;

import org.apache.kafka.clients.producer.Callback;
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.common.record.CompressionType;
import org.apache.kafka.common.serialization.StringSerializer;

import java.util.Properties;
import java.util.function.Consumer;

public class MQTTProducer {

    private KafkaProducer<String, String> producer;

    public MQTTProducer() {
        producer = new KafkaProducer<String, String>(new Properties() {{
            // Kafka服务器集群,逗号分隔
            put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "MatrixEngine:9092");
            // 消息Key序列化方式
            put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
            // 消息Value序列化方式
            put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
            // 缓冲区大小
            put(ProducerConfig.BUFFER_MEMORY_CONFIG, 33554432);
            // 发送队列批次大小
            put(ProducerConfig.BATCH_SIZE_CONFIG, 16384);
            // 如果批次没达到,超过这个时间sender线程也会进行消息发送
            put(ProducerConfig.LINGER_MS_CONFIG, 10);
            // 消息压缩算法(提升I/O速度)
            put(ProducerConfig.COMPRESSION_TYPE_CONFIG, CompressionType.SNAPPY.name);
            // ack级别.
            put(ProducerConfig.ACKS_CONFIG, "all");
            // 重试次数(默认Integer.MAX_VALUE, 此处设置不重试)
            put(ProducerConfig.RETRIES_CONFIG, 0);
            // 开启幂等性
            put(ProducerConfig.ENABLE_IDEMPOTENCE_CONFIG, true);
            // 开启事务(必须配置唯一的transactionId)
            put(ProducerConfig.TRANSACTIONAL_ID_CONFIG, "transaction-0");
            // sender线程对一个连接的Request对象的数量(默认5), 设置为1可以保证消息有序性
            put(ProducerConfig.MAX_IN_FLIGHT_REQUESTS_PER_CONNECTION, 5);
        }});
    }

    public void sendMessage(String topic, String value) {
        producer.send(new ProducerRecord<>(topic, value));
    }

    public void sendMessage(String topic, String key, String value) {
        producer.send(new ProducerRecord<>(topic, key, value));
    }

    public void sendMessage(String topic, Integer partition, String value) {
        producer.send(new ProducerRecord<>(topic, partition, null, value));
    }

    public void sendMessage(String topic, Integer partition, String key, String value) {
        producer.send(new ProducerRecord<>(topic, partition, key, value));
    }

    public void sendMessage(String topic, String value, Callback callback) {
        producer.send(new ProducerRecord<>(topic, value), callback);
    }

    public void sendMessage(String topic, String key, String value, Callback callback) {
        producer.send(new ProducerRecord<>(topic, key, value), callback);
    }

    public void sendMessage(String topic, Integer partition, String value, Callback callback) {
        producer.send(new ProducerRecord<>(topic, partition, null, value), callback);
    }

    public void sendMessage(String topic, Integer partition, String key, String value, Callback callback) {
        producer.send(new ProducerRecord<>(topic, partition, key, value), callback);
    }

    public void sendMessageTransactional(String topic, Integer partition, String key, String value, Callback callback) {
        producer.initTransactions();
        producer.beginTransaction();
        try {
            producer.send(new ProducerRecord<>(topic, partition, key, value), callback);
            producer.commitTransaction();
        } catch (Exception e) {
            producer.abortTransaction();
            e.printStackTrace();
        }
    }

}
